package com.edata.bigdata.util

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.log4j.Logger
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{DataType, DataTypes, StructField, StructType, UserDefinedType}
import org.apache.spark.sql.{DataFrame, Row}

import scala.collection.mutable
import scala.util.matching.Regex


trait Executor extends EDataSparkSession {

  @transient lazy val LOGGER = Logger.getLogger(this.getClass)
  var beanSchema: mutable.HashMap[String, String] = new mutable.HashMap[String, String]()

  def initialize(): Unit

  def findDataFromPath[T](path: String, seperator: String)(implicit bean: Manifest[T]): DataFrame

  def saveDataToPath[T](data: RDD[String], path: String)(implicit bean: Manifest[T]): Unit

  def findData[T](args: String*)(implicit bean: Manifest[T]): DataFrame

  def saveData[T](data: RDD[String], args: String*)(implicit bean: Manifest[T]): Unit

  def updateData[T](data: RDD[String], args: String*)(implicit bean: Manifest[T]): Unit

  def findDataById[T](value: String, args: String*)(implicit bean: Manifest[T]): DataFrame

  def findDataByProperty[T](key: String, value: String, args: String*)(implicit bean: Manifest[T]): DataFrame

  def createDataFrame[T](rdd: RDD[String], seperator: String)(implicit bean: Manifest[T]): DataFrame = {

    val className = bean.toString()
    val schemaJsonStr = beanSchema.getOrElse(className, "")
    if (schemaJsonStr.isEmpty || schemaJsonStr.equals("")) {

    }
    val schemaJsonObj = JSON.parseObject(schemaJsonStr)
    val colsJsonArray = schemaJsonObj.getJSONArray("COLUMNS").toArray
    val structFields = colsJsonArray.map(colJsonObj => {
      val field = colJsonObj.asInstanceOf[JSONObject].getString("COLUMN")
      val dtype = colJsonObj.asInstanceOf[JSONObject].getString("DATATYPE")
      StructField(field, EDataUtils.convertDataType(dtype))
    })

    /*----------------------------------------------------------------------------*/
    def convertDataToRow(data: Array[String], sf: Array[StructField]): Row = {
      val iter = data.iterator
      val row = data.zipWithIndex.map {
        case (element, index) => {
          val dtype = sf(index).dataType
          dtype match {
            case DataTypes.IntegerType => element.toInt
            case DataTypes.StringType => element
            case DataTypes.FloatType => element.toFloat
            case DataTypes.ShortType => element.toShort
            case DataTypes.DoubleType => element.toDouble
            case _ => {
              throw new ClassCastException("Unexpected data type")
            }
          }
        }
      }.toSeq
      Row.fromSeq(row)
    }

    /*---------------------------------------------------------------------------*/
    val data = rdd.map(s => {
      val ss = s.split(seperator)
      convertDataToRow(ss, structFields)
    })
    data.foreach(println)
    SESSION.createDataFrame(data, StructType(structFields))
  }


  def find_SCHEMA(className: String): (String, String, List[String]) = {
    val schemaJsonStr = beanSchema.getOrElse(className, "")
    val schemaJsonObj = JSON.parseObject(schemaJsonStr)
    val label = schemaJsonObj.getJSONObject("OBJECT").getString("TARGET")
    val id = schemaJsonObj.getJSONObject("ID").getString("COLUMN") + ":" + schemaJsonObj.getJSONObject("ID").getString("DATATYPE")
    val col_types = JSON.parseArray(schemaJsonObj.getString("COLUMNS")).toArray.map {
      case obj: JSONObject => obj.getString("COLUMN") + ":" + obj.getString("DATATYPE")
      case _ => ""
    }
    (label, id, col_types.toList)
  }

  def find_GRAPH_SCHEMA(className: String): (String, String, String, String, String, List[String]) = {
    val schemaJsonStr = beanSchema.getOrElse(className, "")
    val schemaJsonObj = JSON.parseObject(schemaJsonStr)
    val space = schemaJsonObj.getJSONObject("GRAPH").getString("SPACE")
    val label = schemaJsonObj.getJSONObject("GRAPH").getString("TARGET")
    val type_ = schemaJsonObj.getJSONObject("GRAPH").getString("TYPE")
    var gid = ""
    if (type_.equals("VERTEX")) {
      gid = schemaJsonObj.getJSONObject("VID").getString("COLUMN") + ":" + schemaJsonObj.getJSONObject("VID").getString("DATATYPE")
    }
    if (type_.equals("EDGE")) {
      val sid = schemaJsonObj.getJSONObject("SID").getString("COLUMN") + ":" + schemaJsonObj.getJSONObject("SID").getString("DATATYPE")
      val tid = schemaJsonObj.getJSONObject("TID").getString("COLUMN") + ":" + schemaJsonObj.getJSONObject("TID").getString("DATATYPE")
      gid = sid + "|" + tid
    }
    val id = schemaJsonObj.getJSONObject("ID").getString("COLUMN") + ":" + schemaJsonObj.getJSONObject("ID").getString("DATATYPE")
    val col_types = JSON.parseArray(schemaJsonObj.getString("COLUMNS")).toArray.map {
      case obj: JSONObject => obj.getString("COLUMN") + ":" + obj.getString("DATATYPE")
      case _ => ""
    }
    (space, label, type_, gid, id, col_types.toList)

  }

}
